package org.tao.lightningmq.nameserver.replication;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.internal.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.event.EventBus;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.common.tcp.TcpMsgDecoder;
import org.tao.lightningmq.common.tcp.TcpMsgEncoder;
import org.tao.lightningmq.nameserver.common.CommonCache;
import org.tao.lightningmq.nameserver.common.MasterSlaveReplicationProperties;
import org.tao.lightningmq.nameserver.common.NameserverProperties;
import org.tao.lightningmq.nameserver.common.TraceReplicationProperties;
import org.tao.lightningmq.nameserver.enums.ReplicationModeEnum;
import org.tao.lightningmq.nameserver.enums.ReplicationRoleEnum;
import org.tao.lightningmq.nameserver.handler.MasterReplicationServerHandler;
import org.tao.lightningmq.nameserver.handler.NodeSendReplicationMsgServerHandler;
import org.tao.lightningmq.nameserver.handler.NodeWriteMsgReplicationServerHandler;
import org.tao.lightningmq.nameserver.handler.SlaveReplicationServerHandler;

import java.util.Objects;

/**
 * @Author lht
 * @date 2025/1/17 - 13:14
 * @description:
 */
public class ReplicationService {

    private static final Logger LOGGER = LoggerFactory.getLogger(ReplicationService.class);

    public ReplicationModeEnum checkProperties() {
        NameserverProperties nameserverProperties = CommonCache.getNameserverProperties();
        String mode = nameserverProperties.getReplicationMode();
        if (StringUtil.isNullOrEmpty(mode)) {
            LOGGER.info("Nameserver 执行单机模式");
            return null;
        }
        ReplicationModeEnum replicationModeEnum = ReplicationModeEnum.getEnumByCode(mode);
        Objects.requireNonNull(replicationModeEnum, "复制模式参数异常");
        if (replicationModeEnum == ReplicationModeEnum.TRACE) {
            // 链路复制模式
            TraceReplicationProperties traceReplicationProperties = nameserverProperties.getTraceReplicationProperties();
            Objects.requireNonNull(traceReplicationProperties.getPort(), "node节点的端口为空");
        } else {
            // 主从复制模式
            MasterSlaveReplicationProperties masterSlaveReplicationProperties = nameserverProperties.getMasterSlaveReplicationProperties();
            assertIsNotBlank(masterSlaveReplicationProperties.getMaster(), "master参数不能为空");
            assertIsNotBlank(masterSlaveReplicationProperties.getRole(), "role参数不能为空");
            assertIsNotBlank(masterSlaveReplicationProperties.getType(), "type参数不能为空");
            Objects.requireNonNull(masterSlaveReplicationProperties.getPort(), "同步端口不能为空");
        }
        return replicationModeEnum;
    }

    public void startReplicationTask(ReplicationModeEnum replicationModeEnum) {
        NameserverProperties nameserverProperties = CommonCache.getNameserverProperties();
        int port = 0;
        ReplicationRoleEnum replicationRoleEnum;
        if (replicationModeEnum == ReplicationModeEnum.MASTER_SLAVE) {
            port = nameserverProperties.getMasterSlaveReplicationProperties().getPort();
            replicationRoleEnum = ReplicationRoleEnum.getEnumByCode(nameserverProperties.getMasterSlaveReplicationProperties().getRole());
        } else {
            port = nameserverProperties.getTraceReplicationProperties().getPort();
            String nextNode = nameserverProperties.getTraceReplicationProperties().getNextNode();
            replicationRoleEnum = StringUtil.isNullOrEmpty(nextNode) ?
                    ReplicationRoleEnum.TAIL_NODE : ReplicationRoleEnum.NODE;
        }

        switch (replicationRoleEnum) {
            case MASTER:
                // master角色，开启netty进程同步数据给master
                startNettyServerAsync(new MasterReplicationServerHandler(new EventBus("master-replication-task-")), port);
                break;
            case SLAVE:
                String masterAddr = nameserverProperties.getMasterSlaveReplicationProperties().getMaster();
                startNettyConnAsync(new SlaveReplicationServerHandler(new EventBus("slave-replication-task-")), masterAddr);
                break;
            case NODE:
                // node角色，开启一个复制数据的异步任务
                String nextNodeAddress = nameserverProperties.getTraceReplicationProperties().getNextNode();
                startNettyServerAsync(new NodeWriteMsgReplicationServerHandler(new EventBus("node-write-msg-replication-task-")), port);
                startNettyConnAsync(new NodeSendReplicationMsgServerHandler(new EventBus("node-send-replication-msg-task-")), nextNodeAddress);
                break;
            case TAIL_NODE:
                // tail node角色，不做处理
                startNettyServerAsync(new NodeWriteMsgReplicationServerHandler(new EventBus("node-write-msg-replication-task-")), port);
                break;
            default:
                throw new RuntimeException("未知的角色");
        }
    }

    /**
     * 开启一个Netty客户端
     * @param inboundHandler 处理器
     * @param address 地址
     */
    private void startNettyConnAsync(SimpleChannelInboundHandler<TcpMsg> inboundHandler, String address) {
        new Thread(() -> {
            EventLoopGroup clientGroup = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            Channel channel;
            bootstrap.group(clientGroup);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new TcpMsgDecoder());
                    ch.pipeline().addLast(new TcpMsgEncoder());
                    ch.pipeline().addLast(inboundHandler);
                }
            });
            ChannelFuture channelFuture = null;
            try {
                Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                    clientGroup.shutdownGracefully();
                    LOGGER.info("nameserver's replication connect application is closed");
                }));
                String[] addr = address.split(":");
                channelFuture = bootstrap.connect(addr[0], Integer.parseInt(addr[1])).sync();
                //连接了master节点的channel对象，建议保存
                channel = channelFuture.channel();
                LOGGER.info("success connected to nameserver replication!");
                CommonCache.setConnectNodeChannel(channel);
                channel.closeFuture().sync();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }).start();
    }

    /**
     * 开启一个Netty服务端
     * @param inboundHandler 处理器
     * @param port 端口
     */
    private void startNettyServerAsync(SimpleChannelInboundHandler<TcpMsg> inboundHandler, int port) {
        new Thread(() -> {
            NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup);
            bootstrap.channel(NioServerSocketChannel.class);
            bootstrap.childHandler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new TcpMsgDecoder());
                    ch.pipeline().addLast(new TcpMsgEncoder());
                    ch.pipeline().addLast(inboundHandler);
                }
            });
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
                LOGGER.info("nameserver's replication application is closed");
            }));
            ChannelFuture channelFuture = null;
            try {
                //master-slave架构
                //写入数据的节点，这里就会开启一个服务
                //非写入数据的节点，这里就需要链接一个服务
                //trace架构
                //又要接收外界数据，又要复制数据给外界
                channelFuture = bootstrap.bind(port).sync();
                LOGGER.info("start nameserver's replication application on port:{}", port);
                //阻塞代码
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }).start();
    }

    private static void assertIsNotBlank(String str, String message) {
        if (StringUtil.isNullOrEmpty(str)) {
            throw new IllegalArgumentException(message);
        }
    }

}
